package cn.com.rocketmq.async;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import java.util.HashSet;
import java.util.Set;

/**
 * 消费者（拉模式）（主动消费）
 * @author xuhaixiang
 * @date 2023-09-10
 */
public class PullConsumer {

    /**
     * 日志对象
     */
    private static  final Logger logger = LoggerFactory.getLogger(PullConsumer.class);

    public static void main(String[] args) {
        try {
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("group_rocketmq_asyncDemo");
            consumer.setNamesrvAddr("192.168.113.101:9876");
            Set<String> topicSet = new HashSet<>();
            topicSet.add("rocketmq_asyncDemo");
            consumer.setRegisterTopics(topicSet);
            consumer.start();
            //主题遍历
            while (true) {
                consumer.getRegisterTopics().forEach(item -> {
                    try {
                        Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues(item);
                        //消息队列
                        messageQueues.forEach(item2 -> {
                            try {
                                long offset = consumer.getOffsetStore().readOffset(item2, ReadOffsetType.READ_FROM_MEMORY);
                                if (offset < 0) {
                                    offset = consumer.getOffsetStore().readOffset(item2, ReadOffsetType.READ_FROM_STORE);
                                }
                                if (offset < 0) {
                                    offset = consumer.maxOffset(item2);
                                }
                                if (offset < 0) {
                                    offset = 0;
                                }
                                PullResult result = consumer.pull(item2, "*", offset, 32);
                                if (result != null) {
                                    switch (result.getPullStatus()) {
                                        case FOUND:{
                                            result.getMsgFoundList().forEach(item3 -> {
                                                try {
                                                    logger.info("消息消费成功！消息ID={}，消息内容：{}", item3.getMsgId(), new String(item3.getBody(), "UTF-8"));
                                                    consumer.updateConsumeOffset(item2, result.getNextBeginOffset());
                                                } catch (Exception e) {
                                                    logger.error("遍历消息信息失败！" , e);
                                                }
                                            });
                                            break;
                                        }
                                        case NO_NEW_MSG:{
                                            logger.info("没有最新消息！");
                                            break;
                                        }
                                        case NO_MATCHED_MSG: {
                                            logger.info("没有匹配的消息！");
                                            break;
                                        }
                                        case OFFSET_ILLEGAL: {
                                            logger.error("偏移量非法，当前偏移量为{}", offset);
                                            break;
                                        }
                                    }
                                }
                            } catch (Exception e) {
                                logger.error("遍历消息队列失败！", e);
                            }
                        });
                    } catch (MQClientException e) {
                        logger.error("遍历主题失败！", e);
                    }
                });
            }
        } catch (MQClientException e) {
            logger.error("消息拉取失败！", e);
        }
    }
}
